package com.atguigu.tingshu.search.mq;

import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.search.service.SearchService;
import com.atguigu.tingshu.vo.album.AlbumStatVoMsg;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;

@Slf4j
public class SearchReceiver {

    @Autowired
    private SearchService searchService;

    // 上架
    @KafkaListener(topics = "tingshu_album_upper")
    public void upperGoods(ConsumerRecord<String, String> consumerRecord){
        String albumId = consumerRecord.value();
        if(albumId != null){
            searchService.upperAlbum(Long.parseLong(albumId));
            log.info("专辑上架成功,albumId:{}" , albumId);
        }
    }

    //下架
    @KafkaListener(topics = "tingshu_album_lower")
    public void lowerGoods(ConsumerRecord<String,String> consumerRecord){
        String albumId = consumerRecord.value();
        if(albumId != null){
            searchService.lowerAlbum(Long.parseLong(albumId));
            log.info("专辑下架成功,albumId:{}" , albumId);
        }
    }
    @KafkaListener(topics = "tingshu.album.stat.update.sync")
    public void updateAlbumStat(ConsumerRecord<String,String> consumerRecord){
        String value = consumerRecord.value();
        if(value != null){
            AlbumStatVoMsg albumStatVoMsg = JSON.parseObject(value, AlbumStatVoMsg.class);
            searchService.updateAlbumStat(albumStatVoMsg);
            log.info("专辑统计数据更新完毕...albumId:{}" , albumStatVoMsg.getAlbumId());
        }
    }
}
